package com.xrui.hbase.codec.impl;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import com.xrui.hbase.AvroSchemaResolver;
import com.xrui.hbase.ColumnName;
import com.xrui.hbase.HBaseURI;
import com.xrui.hbase.codec.CellEncoder;
import com.xrui.hbase.codec.CellEncoderFactory;
import com.xrui.hbase.schema.TableLayout;
import com.xrui.hbase.schema.spec.CellSpec;

import java.io.IOException;
import java.util.Map;
import java.util.Set;

/**
 * Provider for cell encoders of a given table.
 * <p>
 * <p>
 * HBaseCell encoders for all columns in the table are pro-actively created when the
 * CellEncoderProvider is constructed.
 * HBaseCell encoders are cached and reused.
 * </p>
 */
public final class CellEncoderProvider {

    /**
     * Maps column names to encoders.
     */
    private final ImmutableMap<ColumnName, CellEncoder> mEncoderMap;

    /**
     * Initializes a provider for cell encoders.
     *
     * @param tableURI       URI of the table this provider is for.
     * @param layout         the TableLayout for which to provide cell encoders.
     * @param schemaResolver the schema resolver from which to resolve cell schemas.
     * @param factory        the CellEncoderFactory with which to build cell encoders.
     * @throws IOException in case of an error reading from the schema table.
     */
    private CellEncoderProvider(
        final HBaseURI tableURI,
        final TableLayout layout,
        final AvroSchemaResolver schemaResolver,
        final CellEncoderFactory factory)
        throws IOException {

        // Compute the set of all the column names (map-type families and fully-qualified columns from
        // group-type families).
        final Set<ColumnName> columns = layout.getColumns();

        // Pro-actively build cell encoders for all columns in the table:
        final Map<ColumnName, CellEncoder> encoderMap = Maps.newHashMap();
        for (ColumnName column : columns) {
            final CellSpec cellSpec = layout.getCellSpec(column)
                .setColumnURI(
                    HBaseURI.newBuilder(tableURI).withColumnNames(ImmutableList.of(column)).build())
                .setSchemaResolver(schemaResolver)
                .setEncoderFactory(factory);

            final CellEncoder encoder = cellSpec.getEncoderFactory().create(cellSpec);
            encoderMap.put(column, encoder);
        }
        mEncoderMap = ImmutableMap.copyOf(encoderMap);
    }

    /**
     * Initialize a provider for cell decoders.
     *
     * @param tableURI       the table uri for which to provide encoders.
     * @param layout         the layout for which to provide encoders.
     * @param schemaResolver the schema resolver from which to resolve cell schemas.
     * @param factory        Default factory for cell encoders.
     * @return a new {@code CellEncoderProvider}.
     * @throws IOException in case of an error creating the cached encoders.
     */
    public static CellEncoderProvider create(
        final HBaseURI tableURI,
        final TableLayout layout,
        final AvroSchemaResolver schemaResolver,
        final CellEncoderFactory factory
    ) throws IOException {
        return new CellEncoderProvider(tableURI, layout, schemaResolver, factory);
    }

    // -----------------------------------------------------------------------------------------------
    // Public interface

    /**
     * Gets a cell encoder for the specified column or (map-type) family.
     * <p>
     * <p>
     * When requesting a encoder for a column within a map-type family, the encoder for the
     * entire map-type family will be returned.
     * </p>
     *
     * @param column to look up
     * @return a cell encoder for the specified column.
     * Null if the column does not exist or if a family level encoder is requested from a group
     * type family.
     * @throws IOException on I/O error.
     */
    public CellEncoder getEncoder(ColumnName column) throws IOException {
        final CellEncoder encoder = mEncoderMap.get(column);
        if (encoder != null) {
            // There already exists a encoder for this column:
            return encoder;
        }

        if (column.isFullyQualified()) {
            // There is no encoder for the specified fully-qualified column.
            // Try the family (this will only work for map-type families):
            return getEncoder(ColumnName.create(column.getFamily(), null));
        }

        return null;
    }
}
